package com.bw.day0627;


import com.bw.utils.MyKafkaUtil;

import com.bw.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


public class DwdTradeCancelDetail {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        tableEnv.executeSql("" +
                "create table dwd_order_pre( " +
                "    `id` string, " +
                "    `order_id` string, " +
                "    `sku_id` string, " +
                "    `sku_name` string, " +
                "    `order_price` string, " +
                "    `sku_num` string, " +
                "    `create_time` string, " +
                "    `source_type_id` string, " +
                "    `source_type_name` string, " +
                "    `source_id` string, " +
                "    `split_total_amount` string, " +
                "    `split_activity_amount` string, " +
                "    `split_coupon_amount` string, " +
                "    `consignee` string, " +
                "    `consignee_tel` string, " +
                "    `total_amount` string, " +
                "    `order_status` string, " +
                "    `user_id` string, " +
                "    `payment_way` string, " +
                "    `delivery_address` string, " +
                "    `order_comment` string, " +
                "    `out_trade_no` string, " +
                "    `trade_body` string, " +
                "    `operate_time` string, " +
                "    `expire_time` string, " +
                "    `process_status` string, " +
                "    `tracking_no` string, " +
                "    `parent_order_id` string, " +
                "    `province_id` string, " +
                "    `activity_reduce_amount` string, " +
                "    `coupon_reduce_amount` string, " +
                "    `original_total_amount` string, " +
                "    `feight_fee` string, " +
                "    `feight_fee_reduce` string, " +
                "    `refundable_time` string, " +
                "    `order_detail_activity_id` string, " +
                "    `activity_id` string, " +
                "    `activity_rule_id` string, " +
                "    `order_detail_coupon_id` string, " +
                "    `coupon_id` string, " +
                "    `coupon_use_id` string, " +
                "    `type` string, " +
                "    `old` map<string,string>" +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process","order_detail11"));
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        Table table = tableEnv.sqlQuery("select \n" +
                " id, " +
                " order_id, " +
                " user_id, " +
                " sku_id, " +
                " sku_name, " +
                " province_id, " +
                " activity_id, " +
                " activity_rule_id, " +
                " coupon_id, " +
                " create_time," +
                " operate_time ,"+
                " source_id, " +
                " source_type_id, " +
                " source_type_name, " +
                " sku_num, " +
                " order_price, " +
                " split_total_amount, " +
                " split_activity_amount, " +
                " split_coupon_amount " +
                " from dwd_order_pre " +
                "where `type` ='update' " +
                "and `old`['order_status'] is not null " +
                "and order_status = '1003' ");
        tableEnv.createTemporaryView("table1",table);
//        tableEnv.toAppendStream(table, Row.class).print(">>>>>");
    tableEnv.executeSql("create table dwd_trade_cancel_detail(\n" +
            "id                    string," +
            "order_id              string," +
            "user_id               string," +
            "sku_id                string," +
            "sku_name              string," +
            "province_id           string," +
            "activity_id           string," +
            "activity_rule_id      string," +
            "coupon_id             string," +
            "create_time           string," +
            "operate_time          string," +
            "source_id             string," +
            "source_type_id        string," +
            "source_type_name      string," +
            "sku_num               string," +
            "order_price           string," +
            "split_total_amount    string," +
            "split_activity_amount string," +
            "split_coupon_amount   string " +
            ")" +MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cancel_detail"));
    tableEnv.executeSql("insert into dwd_trade_cancel_detail select * from table1");

    }
}
